package cn.ac.iie.bdb_to_mongo.bdbhandler;

import cn.ac.iie.bdb_to_mongo.BdbMain;
import cn.ac.iie.bdb_to_mongo.commons.Constants;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.REMessage;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.TimeUnit;

/**
 * @auther Fighter Created on 2016/12/2.
 */
public class WriteBdbTask implements Runnable {

    private final Logger LOG = LoggerFactory.getLogger(WriteBdbTask.class);

    @Override
    public void run() {
        while (true) {
            try {
                //写bdb,writeFlag为true则从mq拉数据写bdb
                if (BdbMain.writeFlag.get()) {
                    //从队列取出数据新值
                    REMessage message = BdbMain.newValQueue.poll(3, TimeUnit.SECONDS);
                    if (message == null) {
                        LOG.warn("there is no message from the receiver queue!");
                        Thread.sleep(1000);
                        continue;
                    } else {
//                        LOG.info("receive messages num: {}", message.data.size());
                        handle(message);
                    }
                }
            } catch (Exception ex) {
                LOG.error("get data from blockingqueue error! {}", ExceptionUtils.getFullStackTrace(ex));
            }
        }
    }

    private void handle(REMessage message) throws Exception {

        int columnLength = message.colNames.length;//列数
        String[] columnNames = message.colNames;//列名

        for (Object[] data : message.data) {

            String _id = null;
            Long wx_ic = null;

            int dataLength = data.length;
            if (dataLength != columnLength) {
                throw new Exception("data length not equal column length!");
            }

            for (int i = 0; i < columnLength; i++) {
                if (data[i] == null)
                    continue;
                switch (columnNames[i]) {
                    case "_id":
                        _id = data[i].toString();
                        break;
                    case "count":
                        wx_ic = Long.valueOf(data[i].toString());
                        break;
                    default:
                        break;
                }
            }

            boolean putStatus;//写bdb成功或失败标志！
            try {
                putStatus = BdbMain.testDB.put(_id, wx_ic);
                if (!putStatus) {
                    int tryTimes = BdbMain.configuration.getInt(Constants.WRITE_BDB_TRY_TIMES);
                    //重试
                    for (int i = 0; i < tryTimes; i++) {
                        putStatus = BdbMain.testDB.put(_id, wx_ic);
                        if (putStatus)
                            break;
                    }
                }
            } catch (Exception e) {
                LOG.error("WriteBdbTask : write bdb error" + ExceptionUtils.getFullStackTrace(e));
            }

        }
    }
}
